package com.starmark.gateway.log.access.disruptor.publisher;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.starmark.gateway.log.access.disruptor.concurrent.LogThreadFactory;
import com.starmark.gateway.log.access.disruptor.event.LogDataEvent;
import com.starmark.gateway.log.access.disruptor.factory.LogEventFactory;
import com.starmark.gateway.log.access.disruptor.handler.LogDataHandler;
import com.starmark.gateway.log.access.disruptor.translator.LogEventTranslator;
import com.starmark.gateway.log.access.entity.GatewayAccessLog;
import com.starmark.gateway.log.access.service.IGatewayAccessLogService;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * disruptor start and publishEvent.
 *
 * @author starmark
 */
@Component
public class LogEventPublisher implements InitializingBean, DisposableBean {

    private Disruptor<LogDataEvent> disruptor;


    @Value("${log.disruptor.bufferSize:4096}")
    private int bufferSize;

    @Value("${log.disruptor.threadSize:8}")
    private int threadSize;

    @Autowired
    private IGatewayAccessLogService sysAccessLogService;

    /**
     * disruptor start with bufferSize.
     */
    private void start() {
        disruptor = new Disruptor<>(new LogEventFactory(), bufferSize, r -> {
            AtomicInteger index = new AtomicInteger(1);
            return new Thread(null, r, "disruptor-thread-" + index.getAndIncrement());
        }, ProducerType.MULTI, new BlockingWaitStrategy());

        final Executor executor = new ThreadPoolExecutor(threadSize, threadSize, 0, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                LogThreadFactory.create("soul-log-disruptor", false),
                new ThreadPoolExecutor.AbortPolicy());

        LogDataHandler[] consumers = new LogDataHandler[threadSize];
        for (int i = 0; i < threadSize; i++) {
            consumers[i] = new LogDataHandler(executor, sysAccessLogService);
        }
        disruptor.handleEventsWithWorkerPool(consumers);
        disruptor.setDefaultExceptionHandler(new IgnoreExceptionHandler());
        disruptor.start();
    }

    /**
     * publish disruptor event.
     *
     * @param gatewayAccessLog data.
     */
    public void publishEvent(final GatewayAccessLog gatewayAccessLog) {
        final RingBuffer<LogDataEvent> ringBuffer = disruptor.getRingBuffer();
        ringBuffer.publishEvent(new LogEventTranslator(), gatewayAccessLog);
    }

    @Override
    public void destroy() {
        disruptor.shutdown();
    }

    @Override
    public void afterPropertiesSet() {
        start();
    }
}
